package factories

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"hyper/supports/configs"
	"hyper/supports/logger"
	"hyper/supports/queue/abstruct"

	amqp "github.com/rabbitmq/amqp091-go"
)

// 定义 rabbitMQ 结构体
type RabbitMqQueue struct {
	Connection *amqp.Connection
	Ch         *amqp.Channel
	Data       *abstruct.QueueObj
	JsonData   string
}

// 消费者回调方法
type RabbitConsumer func(amqp.Delivery)

func (r *RabbitMqQueue) New(obj *abstruct.QueueObj) *RabbitMqQueue {

	url := fmt.Sprintf(
		"amqp://%s:%s@%s:%s/",
		configs.GetString("queue.rabbitmq.user"),
		configs.GetString("queue.rabbitmq.password"),
		configs.GetString("queue.rabbitmq.host"),
		configs.GetString("queue.rabbitmq.port"))

	// 连接 rabbitMq
	conn, err := amqp.Dial(url)
	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "rabbitMq连接失败",
		})
	}

	// 打开一个通道
	r.Ch, err = conn.Channel()

	// 连接负值方便后期关闭
	r.Connection = conn

	// 队列参数赋值
	r.Data = obj

	return r
}

// 将结构体转为Json
func (r *RabbitMqQueue) MakeJson() *RabbitMqQueue {

	// 获取格式化后得 json 字符串
	jsonStr, err := json.Marshal(r.Data)

	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "转换json 字符串失败",
		})
	}

	r.JsonData = string(jsonStr)

	// 获取队列名称
	r.Data.Name = abstruct.QueueName(r.Data.Name)

	// 声明队列
	_, err = r.Ch.QueueDeclare(
		r.Data.Name, // 队列名称
		true,        // 是否持久化
		false,       // 是否自动删除
		false,       // 是否具有排他性
		false,       // 是否阻塞
		nil,         // 额外属性
	)
	if err != nil {
		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "rabbitMq声明队列失败",
		})
	}

	return r
}

// 压入队列 并返回结果
func (r *RabbitMqQueue) Producer() error {

	// 声明上下文 函数结束时销毁
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
	defer cancel()

	// 压入数据
	err := r.Ch.PublishWithContext(
		ctx,
		"",          // 交换机名称
		r.Data.Name, // 路由键（队列名称）
		false,       // 是否强制
		false,       // 是否立即
		amqp.Publishing{
			DeliveryMode: amqp.Persistent, // 设置持久化
			ContentType:  "text/plain",
			Body:         []byte(r.JsonData),
		})

	// 关闭链接与频道链接
	r.DClose()

	return err
}

// 关闭链接
func (r *RabbitMqQueue) DClose() {
	r.Ch.Close()
	r.Connection.Close()
}

// 消费者代码
func (r *RabbitMqQueue) Consumer(fn RabbitConsumer) {

	// 消费消息
	msgs, err := r.Ch.Consume(
		abstruct.QueueName(r.Data.Name), // 队列名称
		"",                              // 消费者名称
		false,                           // 自动确认（Auto-Ack）
		false,                           // 是否具有排他性
		false,                           // 是否阻塞
		false,                           // 额外参数
		nil,
	)

	if err != nil {

		logger.Error(map[string]interface{}{
			"err":     err.Error(),
			"message": "rabbitMq读取失败",
		})
	}

	var wg sync.WaitGroup
	for d := range msgs {
		wg.Add(1)

		// 处理消息
		fn(d)

	}

	wg.Wait()
}
